package com.atguigu.gmall.item.service.Impl;

import com.atguigu.gmall.common.cache.BloomType;
import com.atguigu.gmall.common.cache.GmallCache;
import com.atguigu.gmall.common.constant.RedisConst;
import com.atguigu.gmall.feign.product.SkuInfoFeignClient;
import com.atguigu.gmall.feign.search.SkuEsFeignClient;
import com.atguigu.gmall.item.service.ItemService;
import com.atguigu.gmall.model.product.BaseCategoryView;
import com.atguigu.gmall.model.product.SkuInfo;
import com.atguigu.gmall.model.product.SpuSaleAttr;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.MappingIterator;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.Maps;
import com.google.common.hash.BloomFilter;
import com.sun.org.apache.bcel.internal.generic.NEW;
import jodd.time.TimeUtil;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.redisson.api.RBloomFilter;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import sun.rmi.runtime.Log;
import sun.util.locale.provider.LocaleServiceProviderPool;

import java.math.BigDecimal;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Slf4j
@Service
public class ItemServiceImpl implements ItemService {

    /*  本地缓存不可行 //临时保存数据
    private Map<String, Object> cache = new HashMap<>();*/

    @Autowired
    SkuInfoFeignClient skuInfoFeignClient;

    @Autowired
    StringRedisTemplate stringRedisTemplate;

    @Autowired
    RedissonClient redissonClient;

    @Qualifier("corePool")
    @Autowired
    ThreadPoolExecutor executor;

    @Autowired
    SkuEsFeignClient esFeignClient;


    /**
     * 1、分布式锁版的获取商品详情
     * 1)、锁有自动过期时间，防止死锁
     * 2）、加锁解锁都是原子的
     * 3）、锁的续期，10s,要给他不断续期，业务部中断，锁要自动延长时间
     * 后台启动一个daemon线程，每隔5秒，让这个锁再续满时间，让锁重新开始倒计时
     * <p>
     * redisson:基于redis的分布式框架
     *
     * @param skuId
     * @return
     */
    @SneakyThrows //使用自己设置的使用原生redis操作实现的分布式可重入锁业务
    public Map<String, Object> getSkuInfoWithRedisLock(Long skuId, Boolean locked, String beforeToken) {

        //1、先判断缓存中是否存在
        Map<String, Object> cache = queryFromCache(skuId);
        if (cache != null) {
            //2、缓存中有用缓存的
            return cache;
        } else {
            log.info("缓存不命中，开始抢锁.........");
            String token = beforeToken;
            Boolean aBoolean = locked;
            if (!locked) {
                token = UUID.randomUUID().toString(); //自己加锁就是新token
                //重入锁的设计
                aBoolean = stringRedisTemplate.opsForValue().setIfAbsent("lock", beforeToken, 20, TimeUnit.SECONDS);
            }
            //SETNX
            //3、缓存中没有调用业务逻辑真正查询
            //3.1 占锁
            if (aBoolean) {
                try {
/*                    //5、删除锁
                    String lock = stringRedisTemplate.opsForValue().get("lock");
                    if (token.equals(lock)) {
                        stringRedisTemplate.delete("lock");
                        log.info("释放锁..................");
                    }*/
                    log.info("抢锁成功----------");
                    cache = getFromServiceItemFeign(skuId);
                    saveToCache(cache, skuId);
                } finally {
                    //5、原子删除锁
                    String script = "if redis.call('get', KEYS[1]) == ARGV[1]" +
                            " then return redis.call('del', KEYS[1]) else return 0 end";
                    stringRedisTemplate.execute(new DefaultRedisScript<>(script), Arrays.asList("lock"), beforeToken);
                    log.info("释放锁..................");

                    /*  String lock = stringRedisTemplate.opsForValue().get("lock");
                    if (token.equals(lock)) {
                        stringRedisTemplate.delete("lock");
                    }*/
                }
            } else {
                //3.3 站锁失败,没有占锁成功就一直自旋
                log.info("没抢成功，开始尝试自旋占锁.................");
                while (true) {
                    log.info("自旋中.........");//同一线程的锁应该重复使用
                    try {
                        Thread.sleep(200);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    String token2 = UUID.randomUUID().toString();
                    Boolean absent = stringRedisTemplate.opsForValue().setIfAbsent("lock", token2, 20, TimeUnit.SECONDS);
                    if (absent) {
                        log.info("自旋占锁成功...............................................");
                        cache = getSkuInfoWithRedisLock(skuId, absent, token2);
                        //释放锁
                        String script = "if redis.call('get', KEYS[1]) == ARGV[1]" +
                                " then return redis.call('del', KEYS[1]) else return 0 end";
                        DefaultRedisScript<Object> defaultRedisScript = new DefaultRedisScript<>(script);
                        stringRedisTemplate.execute(defaultRedisScript, Arrays.asList("lock"), token);
                        log.info("释放锁..................");
                        return cache;
                    }
                }
            }
        }
        return cache;
    }


    @SneakyThrows
    public Map<String, Object> queryFromCache(Long skuId) {
        ObjectMapper mapper = new ObjectMapper();
        ValueOperations<String, String> operations = stringRedisTemplate.opsForValue();
        String redisContent = operations.get("sku:info" + skuId);
        if (StringUtils.isEmpty(redisContent)) {
            return null;
        } else {
            log.info("缓存命中...............");
            Map<String, Object> stringObjectMap = mapper.readValue(redisContent, new TypeReference<Map<String, Object>>() {
            });
            return stringObjectMap;
        }

    }

    @SneakyThrows
    void saveToCache(Map<String, Object> feign, Long skuId) {
        //null值也要缓存
        ObjectMapper objectMapper = new ObjectMapper();
        String string = objectMapper.writeValueAsString(feign);

        ValueOperations<String, String> operations = stringRedisTemplate.opsForValue();
        operations.set("sku:info" + skuId, string);
    }

    /**
     * 分布式锁version02
     * 1、redisson会死锁嘛？
     * 不会，正常编码
     * 1、redisson 是可重入锁（同一个线程内前面加过后面不用加）
     * 2、redisson 会自动解锁嘛，
     * 会，断电看门狗过30s，代码闪断，定时线程没有了，就不自动续期了，等30s后redis自己删除
     * 3、不推荐使用传时间，看门狗自动续期失效，底层是判断leaseTime是否有值，有值就没有续期功能
     * <p>
     * 2、redisson分布式业务超时怎么办？
     * 1、看门狗自动续期，每隔1/3看门狗时间就续满看门狗时间
     * <p>
     * <p>
     * 3、缓存的一致性问题
     * 数据库修改以后，缓存的数据需要同步过来
     * 1）、sku:info:50 :后台改了50号数据，30min分钟过期（允许有一段时间数据是不一致的）【最终一致性】
     * 数据过期以后，下一次对此数据的查询会触发更新逻辑，最终缓存中还是新数据
     * <p>
     * 【双写模式】：本来要同步缓存（改数据的同时给缓存存一份）；写数据库+写缓存
     * 【失效模式】：写数据库+删缓存：下一次查询，由于缓存中没有，自己会查数据库，然后又更新到缓存
     * 这两种方法都可以，99%都是运行成功的，如果失败，我们可以寄希望与30min过期
     * <p>
     * 【一劳永逸】：把数据库的最后一次修改存入缓存
     * 【Canal】更新缓存，该数据库同步改缓存，
     * 安装Canal:开发同步到binLog以后的代码，sku_info
     * sku_info_update_5:  拿到5号记录放到缓存
     * 【mq】:
     * 业务操作改了数据库，给mq发消息。
     * id:5    value:xxxx  updateTime:获取到数据库的修改时间  发给mq
     * 有线程专门订阅mq消息，把消息放到缓存中，mq消息乱序
     * 缓存中此记录的时间如果大于我接下来要给缓存中放的数据时间。我就放弃此操作
     * <p>
     * 总结：一、什么是缓存一致性问题，如何解决的？
     * 1、我们缓存数据是实时性一致性不高的（可以容忍一段时间不一致）
     * 2、缓存一致性问题，缓存中的数据没有同步到数据库最新的数据
     * 3、我们业务代码是，失效模式+缓存数据过期时间。
     * 4、我们失效模式导致了失败与错误，会在一段时间后数据过期，下一次查询出发查询最新的数据
     * 二、怎么保证大并发下，缓存没有，查到的数据也是能放到缓存？怎么解决缓存击穿问题，缓存雪崩问题
     * 1）、加锁
     * 分布式锁性能太低了
     * 直接操作4000
     * 缓存没有要正分布式锁：50/s
     * 缓存中有：600/s
     * 最终决定：只加本地锁
     * <p>
     * 4、锁的粒度问题：
     * 锁粒度越细，越快
     * （正常情况锁方法，那么会把全部的sku并发锁住，如果给锁拼接个skuId加以区分，就减小了锁的粒度）
     * <p>
     * 5、分布式数据一致性问题
     * raft:(强一致性)
     * 1，2，3
     * A-------1------把x号记录改成了y----返回成功或者失败
     * 返回成功：数据库集群大多数节点都已经同步了就会给你说成功
     * B-------2------获取x号记录----领导会告诉还没有记录
     * B------2---获取x记录，领导会返回记录
     * mysql:后台同步
     * 写：连上master:
     * master上改了东西一定不是立即读出来
     * 读：连上slaver
     * 大概再1s左右读来
     * mysql：
     * 1、cpu在1s的时候给mysql提交保存数据
     * 2、mysql需要三秒才保存完
     * 3、cpu在2s的时候要mysal此条数据
     * 4、mysql肯定返回没有
     * <p>
     * 缓存：失效模式+过期时间+本地所
     * 缓存穿透攻击：
     * 差一个不存在的值，缓存没有总查数据库，允许缓存null值；
     * 随机值攻击：（缓存穿透攻击）
     * 1）/api/item/50 1~10000
     * 2)/api/item/时间戳变成数字
     * 核心思想：缓存中有，基本上就是数据库有的。
     * 布隆过滤器:
     * 1)准备一个布隆过滤器
     * 1.1）布隆说有，不一定有
     * 查缓存，查数据库，缓存null[利用布隆说有的进行null攻击]，请求达不到数据库
     * 1.2）布隆说没有就一定没有，直接返回null，缓存都不用看
     * 利用占位（bit位“0010101000”）进行三次（多次）哈希计算，一个key分别占用不用的bit位，
     * 误判率越小，布隆的 bit位数可能会最大    hash时间复杂度是O(1)
     * 在分布式系统中，布隆过滤器中间件的大布隆，并且定期重置
     *
     * @param skuId
     * @return
     */
    //基于redisson分布式锁做的方法
    public Map<String, Object> getSkuInfoWithRedisson(Long skuId) {

        //log.info("准备查询skuId数据:{}",skuId+"号数据");
        /*
        原生方式
         */
        //return getSkuInfoWithRedisLock(skuId,false,null);


        //2、redisson版
        RLock lock = redissonClient.getLock("lock");
        //看缓存
        Map<String, Object> cache = queryFromCache(skuId);
        if (cache == null) {
            log.info("redis缓存没命中，准备查询数据库");
            try {
                log.info("加锁..........");
                lock.lock();
                log.info("加锁成功，尝试继续命中缓存.....");
                Map<String, Object> cacheAgain = queryFromCache(skuId);
                if (cacheAgain == null) {
                    Map<String, Object> data = getFromServiceItemFeign(skuId);
                    log.info("【查询数据库:{}】", data);
                    saveToCache(data, skuId);
                    return data;
                }
                log.info("命中缓存，直接返回.........");
                return cacheAgain;
            } finally {
                lock.unlock();
            }
        }
        log.info("缓存命中，直接返回-=-=-=-=-=-=-");
        return cache;
    }

    /**
     * 加注解后，会先执行缓存注解中的方法，可以进行判断布隆加缓存等等
     *
     * @param skuId
     * @return
     */
    @GmallCache(bloomPrefix = RedisConst.SKUKEY_PREFIX,
            bloomSuffix = RedisConst.SKUKEY_SUFFIX,
            cacheKeyExpr = RedisConst.SKUKEY_PREFIX +
                    "#{#args[0]}" +
                    RedisConst.SKUKEY_SUFFIX,
            ttl = 1000 * 60 * 30,
            bloomName = BloomType.SKU,
            missTtl = 1000 * 60 * 10
    )
    @Override
    public Map<String, Object> getSkuInfo(Long skuId) {
        //查数据库
        log.info("查数据库了..........");
        Map<String, Object> map = getFromServiceItemFeign(skuId);
        return map;
    }

    /**
     * 更新商品skuId的热度
     *
     * @param skuId
     */
    @Override
    public void updateSkuHotScore(Long skuId) {
        //每次点击商品都更新热度
        //100w 100 当redis skuId: 200
        CompletableFuture.runAsync(()->{
            ZSetOperations<String, String> zset = stringRedisTemplate.opsForZSet();
            Double hotScore = zset.incrementScore("hotScore", skuId.toString(), 1);
            if (hotScore % 10 == 0) {
                esFeignClient.updateSkuHotScore(skuId, hotScore.longValue());
            }
        },executor);
    }

    /**
     * 远程调用service-product 查询出和当前skuId 对应的所有信息
     *
     * @param skuId
     * @return
     *//*
    @Override
    public Map<String, Object> getSkuInfo(Long skuId) {

        ObjectMapper mapper = new ObjectMapper();

*//*      本地缓存不可行  if (cache.get("data") != null){
            //缓存中有
            return (Map<String, Object>) cache.get("data");
        }else {
            cache.put("data",result);
        }*//*

        ValueOperations<String, String> operations = stringRedisTemplate.opsForValue();
        //1、先去缓存确定是否存在
        String s = operations.get("sku:info" + skuId);
        if (StringUtils.isEmpty(s)) {
            //2、缓存中没有
            Map<String, Object> fromServiceItemFeign = getFromServiceItemFeign(skuId);

            String jsonStr = null;
            try {
                jsonStr = mapper.writeValueAsString(fromServiceItemFeign);
            } catch (JsonProcessingException e) {
                e.printStackTrace();
            }
            operations.set("sku:info" + skuId, jsonStr);
            return fromServiceItemFeign;
        } else {
            Map<String, Object> stringObjectMap = null;
            try {
                stringObjectMap = mapper.readValue(s, new TypeReference<Map<String, Object>>() {});
            } catch (JsonProcessingException e) {
                e.printStackTrace();
            }
            return stringObjectMap;
        }
    }*/
    private Map<String, Object> getFromServiceItemFeign01(Long skuId) {
        log.info("开始操作数据库，远程调用");

        /**
         * 1、Sku基本信息
         * 2、Sku图片信息
         */
        SkuInfo skuInfo = skuInfoFeignClient.getSkuInfo(skuId);

        /**
         * 7、页面
         */
        Map<String, Object> result = new HashMap<>();
        if (skuInfo != null) {
            result.put("skuInfo", skuInfo);
            //只要操作数据库，就
            /**
             * 3、Sku分类信息
             */
            BaseCategoryView skuCategorys = skuInfoFeignClient.getSkuCategorys(skuInfo.getCategory3Id());
            result.put("categoryView", skuCategorys);
            /**
             * 4、Sku销售属性相关信息
             */
            List<SpuSaleAttr> spuSaleAttrListCheckBySku = skuInfoFeignClient.getSpuSaleAttrListCheckBySku(skuId, skuInfo.getSpuId());
            result.put("spuSaleAttrList", spuSaleAttrListCheckBySku);
            /**
             * 5、Sku的价格
             */
            BigDecimal skuPrice = skuInfoFeignClient.getSkuPrice(skuId);
            result.put("price", skuPrice);
            /**
             * 6、SPU下面的所有存在的sku组合信息
             * TODO 可能要转json字符串
             */
            Map map = skuInfoFeignClient.getSkuValueIdsMap(skuInfo.getSpuId());
            ObjectMapper mapper = new ObjectMapper();
            String jsonStr = null;
            try {
                jsonStr = mapper.writeValueAsString(map);
            } catch (JsonProcessingException e) {
                log.error("商品sku组合数据转换异常：{}", e);
            }
            result.put("valuesSkuJson", jsonStr);
        }
        return result;
    }

    /**
     * 异步处理查询
     * 优点：
     * 线程的重复使用
     * 线程吞吐量的限制
     *
     * @param skuId
     * @return
     */
    private Map<String, Object> getFromServiceItemFeign(Long skuId) {
        log.info("异步处理查询数据库，skuId = {}", skuId);
        HashMap<String, Object> result = Maps.newHashMap();

        CompletableFuture<SkuInfo> future = CompletableFuture.supplyAsync(() -> {
            SkuInfo skuInfo = skuInfoFeignClient.getSkuInfo(skuId);
            result.put("skuInfo", skuInfo);
            return skuInfo;
        }, executor);


        CompletableFuture<Void> future1 = future.thenAcceptAsync((res) -> {
            if (res != null) {
                BaseCategoryView skuCategorys = skuInfoFeignClient.getSkuCategorys(res.getCategory3Id());
                result.put("categoryView", skuCategorys);
            }
        }, executor);

        CompletableFuture<Void> future2 = future.thenAcceptAsync((res) -> {
            if (res != null) {
                List<SpuSaleAttr> spuSaleAttrListCheckBySku = skuInfoFeignClient.getSpuSaleAttrListCheckBySku(skuId, res.getSpuId());
                result.put("spuSaleAttrList", spuSaleAttrListCheckBySku);
            }
        }, executor);

        CompletableFuture<Void> future3 = future.thenAcceptAsync((res) -> {
            if (res != null) {
                BigDecimal skuPrice = skuInfoFeignClient.getSkuPrice(skuId);
                result.put("price", skuPrice);
            }
        }, executor);

        CompletableFuture<Void> future4 = future.thenAcceptAsync((res) -> {
            if (res != null) {
                Map map = skuInfoFeignClient.getSkuValueIdsMap(res.getSpuId());
                ObjectMapper mapper = new ObjectMapper();
                try {
                    String jsonStr = mapper.writeValueAsString(map);
                    log.info("valuesSkuJson内容:{}" + jsonStr);
                    if (jsonStr != null) {
                        result.put("valuesSkuJson", jsonStr);
                    }
                } catch (JsonProcessingException e) {
                    log.error("商品sku组合数据转换异常：{}", e);
                }
            }
        }, executor);

        CompletableFuture<Void> allOf = CompletableFuture.allOf(future, future1, future2, future3, future4);
        try {
            allOf.get();//阻塞所有的完成再结束
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            log.error("线程池异常:{0}", e);
        }
        return result;
    }
}
